package org.example.process;

import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.example.data.WaterSensor;
import org.example.function.WaterSensorMapFunction;

import java.time.Duration;

public class KeyedProcessTimerDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> source = env.socketTextStream("localhost", 9999);
        SingleOutputStreamOperator<WaterSensor> waterSensorStream = source.map(new WaterSensorMapFunction());

        WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy
                //指定乱序的Watermark生成，延时。
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((WaterSensor waterSensor, long l) -> {
                    //返回的时间戳 毫秒
                    return waterSensor.getTs() * 1000;
                });

        SingleOutputStreamOperator<WaterSensor> waterSensorDsWithWatermark = waterSensorStream.assignTimestampsAndWatermarks(waterSensorWatermarkStrategy);
        KeyedStream<WaterSensor, String> waterSensorStringKeyedStream = waterSensorDsWithWatermark.keyBy(WaterSensor::getId);

        SingleOutputStreamOperator<String> process = waterSensorStringKeyedStream.process(new KeyedProcessFunction<String, WaterSensor, String>() {
            //来一条数据处理一次
            @Override
            public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                //获取数据中提取出来的事件时间，如果没有为null。
                Long currentEventTs = ctx.timestamp();
                String currentKey = ctx.getCurrentKey();
                //事件时间定时器
                TimerService timerService = ctx.timerService();
                timerService.registerEventTimeTimer(5000L);

                //处理时间定义器
                //long currentProcessingTime = timerService.currentProcessingTime();
                //timerService.registerProcessingTimeTimer(currentProcessingTime + 5000L);

                System.out.println("当前Key=" + currentKey + "当前事件时间为 = " + currentEventTs + ", 注册了一个5s的定时器");

                //获取当前的系统时间
                Long currentTs = timerService.currentProcessingTime();

                Long ws = timerService.currentWatermark();
            }

            //时间进展到定义器设定的时间时，调用该方法

            /**
             *
             * @param timestamp 当前的时间进展
             * @param ctx 上下文
             * @param out 采集器
             * @throws Exception
             */
            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                super.onTimer(timestamp, ctx, out);
                String currentKey = ctx.getCurrentKey();
                System.out.println("当前Key=" + currentKey + "当前时间是 = " + timestamp + ", 由定时器触发 ");
            }
        });

        process.print();

        env.execute("PartitionDemo");
    }
}
